package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/8
 *
 *      一个流的数据和另一个在指定的时间间隔内到达的流的数据，可以进行关联。
 *      无需进行窗口运算。
 *      必须keyBy。
 *      默认按照keyBy(key)进行关联。
 */
public class Demo8_IntervalJoin
{
    public static void main(String[] args) throws Exception {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs())
             ;

        //两个流
        KeyedStream<WaterSensor, String> ds1 = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);

        KeyedStream<WaterSensor, String> ds2 = env
            .socketTextStream("hadoop102", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);



        /*
            左侧流中的每一条数据都会和右侧流中指定时间范围到达的数据进行关联。

            迟到数据无法关联！
                何谓迟到:  当前数据的 时间属性 < process的watermark
         */
        ds1.intervalJoin(ds2)
           //默认只支持inEventTime
           //.inProcessingTime()
           //.inEventTime()
           // 以当前数据的时间为0，正数，就是未来时间，负数就是过去时间
           .between(Time.seconds(-3),Time.seconds(3))
           // 数据卡点，是否也关联。默认关联
           //.upperBoundExclusive()
           //.lowerBoundExclusive()
           .process(new ProcessJoinFunction<WaterSensor, WaterSensor, String>()
           {
               @Override
               public void processElement(WaterSensor left, WaterSensor right, Context ctx, Collector<String> out) throws Exception {

                   out.collect(left +"===" +right);

               }
           })
           .print();

        env.execute();

    }
}
